

#include "muduo/prot/codec.h"
#include "muduo/prot/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/TcpConnection.h"

#include "mq_worker.hpp"
#include "mq_channel.hpp"

namespace xgmq
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string &ip, int port, const AsyncWorker::ptr &worker)
            : _worker(worker),
              _client(_worker->loopthread.startLoop(), muduo::net::InetAddress("127.0.0.1", 8085), "client"),
              _dispatcher(std::bind(&Connection::onUnknownMessage, this, std::placeholders::_1,
                                    std::placeholders::_2, std::placeholders::_3)),
              _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher,
                                                               std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
              _latch(1),
              _channel_manager(std::make_shared<ChannelManager>())
        {
            // 给分发器注册方法,这里需要指明类型
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::OnCommonResponse, this, std::placeholders::_1,
                                                                               std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::OnBasicResponse, this, std::placeholders::_1,
                                                                                std::placeholders::_2, std::placeholders::_3));

            // 给服务器设置回调函数
            // 收到消息后的回调函数
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1,
                                                 std::placeholders::_2, std::placeholders::_3));
            // 连接变化的回调函数
            _client.setConnectionCallback(std::bind(&Connection::OnConnection, this, std::placeholders::_1));

            _client.connect(); // 与服务器建立连接
            _latch.wait();     // 在这里阻塞，直到建立连接成功
        }
        // 创建一个信道
        Channel::ptr createChannel()
        {
            // 在这里客户端创建一个信道，也要通知服务器端创建一个信道
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            channel->openChannel();
            return channel;
        }
        // 关闭一个信道
        void closeChannel(const Channel::ptr &channel)
        {
            channel->closeChannel();
        }

    private:
        // 通用响应的回调函数
        void OnCommonResponse(const muduo::net::TcpConnectionPtr &conn,
                              const basicCommonResponsePtr &message,
                              muduo::Timestamp)
        {
            // 1，找到对应的信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("信道不存在\n");
                return;
            }
            // 2,将该响应放入信道的哈希表中
            channel->putBasicResponse(message);
        }
        // 消息推送回调函数
        void OnBasicResponse(const muduo::net::TcpConnectionPtr &conn,
                             const basicConsumeResponsePtr &message,
                             muduo::Timestamp)
        {
            // 1，找到对应的信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("信道不存在\n");
                return;
            }
            // 2,封装成任务抛入线程池中
            _worker->pool.push([channel,message]()
                               { channel->consume(message); });
        }

        // 新连接建立成功的回调函数
        void OnConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                _conn = conn;
                _latch.countDown(); // 唤醒阻塞的主线程
            }
            else
            {
                // 重置连接
                _conn.reset();
            }
        }
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                              const MessagePtr &message,
                              muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

    private:
        AsyncWorker::ptr _worker;
        muduo::net::TcpConnectionPtr _conn; // 连接管理
        muduo::net::TcpClient _client;      // 客户端
        muduo::CountDownLatch _latch;       // 实现同步
        ProtobufDispatcher _dispatcher;     // 请求分发器对象，每一种请求对应一个回调方法
        ProtobufCodecPtr _codec;            // protobuf协议处理器，将收到的消息通过协议 进行序列化和反序列化
        // 再通过分发器完成业务层面的处理
        ChannelManager::ptr _channel_manager;
    };
}